package com.kim.client.websocket;

import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.toolkit.StringUtils;
import com.kim.client.entity.SessionInfo;
import com.kim.client.enums.MessageTypeServerEnum;
import com.kim.client.factory.StrategyBeanFactory;
import com.kim.client.interfaces.MessageStrategy;
import com.kim.common.constant.ImRedisKey;
import com.kim.common.entity.cache.TerminalConnectInfo;
import com.kim.common.entity.cache.UserTerminalConnectCache;
import com.kim.common.entity.dto.TokenUserInfo;
import com.kim.common.entity.message.MessageInfo;
import com.kim.common.enums.GlobalExceptionEnum;
import com.kim.common.thread.Worker;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.inject.Inject;
import jakarta.websocket.*;
import jakarta.websocket.server.PathParam;

import jakarta.websocket.server.ServerEndpoint;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RedissonClient;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;


@ServerEndpoint("/ws/chat/{token}/{terminalType}")
@Slf4j
@ApplicationScoped
public class WebSocketServer {


    //    @Inject
//    @Getter
//    Session session;

    /**
     * 连接sid和连接会话
     */
    private String token;

    public ConcurrentHashMap<String, UserTerminalConnectCache> localUserConnectCache = new ConcurrentHashMap<>();
    /**
     * save sessionId -> uid
     */
    public ConcurrentHashMap<String, String> sessionToUid = new ConcurrentHashMap<>();

    /**
     * 本地用户终端连接信息 uid -> sessionInfo
     */
    public ConcurrentHashMap<String, SessionInfo> localTerminalConnectInfo = new ConcurrentHashMap<>();


    //    @Value("${netty.server.port}")
    private String port;

    @Inject
    StrategyBeanFactory factory;


    @Inject
    RedissonClient redissonClient;


    /**
     * 终端连接
     *
     * @param token
     * @param terminalType 终端类型(ios android pc web )
     * @param session
     */
    @OnOpen
    public void onOpen(@PathParam("token") String token, @PathParam("terminalType") Integer terminalType, Session session) {
        try {

            log.info("token:{}", token);
            // first obtain user information based on the token from redis(先从redis中根据token获取用户信息)
            Object o = redissonClient.getBucket(ImRedisKey.TOKEN_USER_INFO + token).get();
            TokenUserInfo tokenUserInfo = JSONObject.parseObject((String) o, TokenUserInfo.class);

            UserTerminalConnectCache userTerminalConnectCache = new UserTerminalConnectCache();
            // set redis user connect info(设置全局redis用户连接信息)
            userTerminalConnectCache = setUserClientConnectCache(terminalType, session, tokenUserInfo, userTerminalConnectCache);
            //  you need to save the token information,user information and connect client information to the global cache
            //  (需要将token信息、用户信息  以及连接的terminal信息保存到全局缓存)
            redissonClient.getBucket(ImRedisKey.USER_CONNECT_INFO + tokenUserInfo.getUid()).set(userTerminalConnectCache);
            UserTerminalConnectCache localUserConnectCache = this.localUserConnectCache.get(tokenUserInfo.getUid());
            // set local user connect info(设置本地缓存用户连接信息)
            if (localUserConnectCache == null) {
                this.localUserConnectCache.put(tokenUserInfo.getUid(), userTerminalConnectCache);
            } else {
                TerminalConnectInfo terminalConnectInfo = setTerminalConnectInfo(terminalType, session, tokenUserInfo);
                localUserConnectCache.getTerminalConnectInfos().add(terminalConnectInfo);
            }

        } catch (UnknownHostException e) {
            throw new RuntimeException(e);
        }
        SessionInfo info = SessionInfo.builder().session(session).build();
        localTerminalConnectInfo.put(session.getId(), info);
        WebSocketManager.sentToUser(session, "WebSocket is connected!");
        WebSocketManager.addWebSocketServer(session);
        log.info("与SessionId：{}created connect(建立连接)", session.getId());
    }

    /**
     * 设置客户连接缓存
     *
     * @param terminalType             终端类型
     * @param session                  会话信息
     * @param tokenUserInfo            用户信息
     * @param userTerminalConnectCache 用户客户端连接缓存
     * @return UserClientConnectCache 用户客户端连接缓存
     * @throws UnknownHostException
     */
    private UserTerminalConnectCache setUserClientConnectCache(Integer terminalType, Session session, TokenUserInfo tokenUserInfo, UserTerminalConnectCache userTerminalConnectCache) throws UnknownHostException {
        String s = ImRedisKey.USER_CONNECT_INFO + tokenUserInfo.getUid();
        if (StringUtils.isNotBlank(s)) {
            userTerminalConnectCache = JSONObject.parseObject(s, UserTerminalConnectCache.class);
            userTerminalConnectCache.getTerminalConnectInfos().add(setTerminalConnectInfo(terminalType, session, tokenUserInfo));
        } else {

            TerminalConnectInfo terminalConnectInfo = setTerminalConnectInfo(terminalType, session, tokenUserInfo);
            userTerminalConnectCache.setUserId(userTerminalConnectCache.getUserId());
            LinkedList<TerminalConnectInfo> objects = new LinkedList<>();
            objects.add(terminalConnectInfo);
            userTerminalConnectCache.setTerminalConnectInfos(objects);
        }
        return userTerminalConnectCache;
    }

    /**
     * 设置终端连接信息
     *
     * @param terminalType
     * @param session
     * @param tokenUserInfo
     * @return
     * @throws UnknownHostException
     */
    private TerminalConnectInfo setTerminalConnectInfo(Integer terminalType, Session session, TokenUserInfo tokenUserInfo) throws UnknownHostException {
        InetAddress localHost = InetAddress.getLocalHost();

        return TerminalConnectInfo.builder()
                .terminalType(terminalType).sessionId(session.getId())
                .nettyClientIp(localHost.getHostAddress()).port(port)
                .build();
    }

    @OnClose
    public void onClose(Session session) {

        // obtain the uid based on the sessionId (根据sessionId 获取uid)
        String uid = sessionToUid.get(session.getId());
        // deletes the terminal connection information in the local cache based on the uid(根据uid删除本地缓存中的终端连接信息)
        UserTerminalConnectCache userTerminalConnectCache = localUserConnectCache.get(uid);
        LinkedList<TerminalConnectInfo> terminalConnectInfos = userTerminalConnectCache.getTerminalConnectInfos();
        Iterator<TerminalConnectInfo> iterator = terminalConnectInfos.iterator();
        while (iterator.hasNext()) {
            /**
             * 如果当前用户只有一个终端连接 且下线的终端sessionId就是目前在线的唯一终端 说明本用户目前使用的终端下线后  该用户没有在线的终端了  删除本地和全局缓存
             */
            if (terminalConnectInfos.size() == 1) {
                if (terminalConnectInfos.get(0).getSessionId().equals(session.getId())) {
                    localUserConnectCache.remove(uid);
                    localTerminalConnectInfo.remove(terminalConnectInfos.get(0).getSessionId());
                    redissonClient.getBucket(ImRedisKey.USER_CONNECT_INFO + uid).delete();
                }
            } else {
                // update the local and global caches(更新本地缓存和全局缓存)
                if (iterator.next().getSessionId().equals(session.getId())) {
                    iterator.remove();
                    // update the terminal connection information in the global cache based on the uid(根据uid更新全局缓存中的终端连接信息)
                    redissonClient.getBucket(ImRedisKey.USER_CONNECT_INFO + uid).set(userTerminalConnectCache);
                    break;
                }
            }
        }
        // delete local cache (删除本地缓存)
        sessionToUid.remove(session.getId());
        localTerminalConnectInfo.remove(session.getId());
        WebSocketManager.removeWebSocketServer(session);
        log.info("WebSocket closed connect(连接关闭)");
    }

    @OnMessage
    public void onMessage(String message, Session session) {
        log.info("来自SessionId:{}的消息:{}", session.getId(), message);
        MessageInfo messageInfo = JSONObject.parseObject(message, MessageInfo.class);
        if (messageInfo.getMessageType() != null) {
            log.error("消息格式错误，messageType 不能为空");
            WebSocketManager.sentToUser(session, "消息格式错误，messageType 不能为空");
//            WebSocketManager.addWebSocketServer(session);
        }

        // 线程池执行
//        TaskToolExecutor executor = executorMap.get("msgExecutor");
//        final CountDownLatch countDownLatch = new CountDownLatch(1);
//        Worker<Object> worker = createSendMessageWorker(messageInfo, session, countDownLatch);
//        try {
//            countDownLatch.await();
//        } catch (InterruptedException e) {
//            log.error("线程池异常：{}------{}", e.getMessage(), e.getStackTrace());
//            countDownLatch.countDown();
//        }
//
//        Object result = executor.submit(worker);
//        WebSocketManager.sentToUser(session, String.valueOf(result));

        if (StringUtils.isNotBlank(messageInfo.getMessageType())) {
            MessageTypeServerEnum messageStrategyEnum = MessageTypeServerEnum.getServiceById(messageInfo.getMessageType());
            if (messageStrategyEnum != null) {
                try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
                    Future<GlobalExceptionEnum> future = (Future<GlobalExceptionEnum>) executor.submit(() -> {
                        MessageStrategy processor = factory.getProcessor(messageStrategyEnum.getService());
                        processor.onMessage(messageInfo, session);
                        log.info("send message success");
                    });
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }


            }
        } else {

            log.error("消息格式错误，messageInfo.MessageType 不能为空！");
        }


    }

    private Worker<Object> createSendMessageWorker(MessageInfo messageInfo, Session session, CountDownLatch countDownLatch) {
        Worker<Object> worker = new Worker<>();
        Runnable runnable = () -> {


            if (StringUtils.isNotBlank(messageInfo.getMessageType())) {
                MessageTypeServerEnum messageStrategyEnum = MessageTypeServerEnum.getServiceById(messageInfo.getMessageType());
                if (messageStrategyEnum != null) {
                    try {
                        MessageStrategy processor = factory.getProcessor(messageStrategyEnum.getService());
                        processor.onMessage(messageInfo, session);
                        worker.setResult("send message success");
                    } catch (Exception e) {
                        // 设置返回值
                        worker.setResult(e.getMessage());

                    }
                }
                countDownLatch.countDown();
            } else {
                log.error("消息格式错误，messageInfo.MessageType 不能为空！");
            }
        };
        worker.setTimeout(500);
        worker.setCommand(runnable);
        return worker;
    }

    @OnError
    public void onError(Session session, Throwable error) {
        log.error("Session:{}的WebSocket发生错误", session.getId(), error);
    }


}